package com.shujia.spark.sql

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}

object Demo2SQL {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("app")

    //设置sparksql  默认分区数，   在产生shuffle过程中会起作用
    conf.set("spark.sql.shuffle.partitions", "1")

    val sc = new SparkContext(conf)

    // spark  sql 上下文对象  spark sql的入口
    val sqlContext = new SQLContext(sc)

    val students: DataFrame = sqlContext.read.json("spark/data/students.json")

    students.show(100)

    //注册一张表，就可以写sql了
    students.registerTempTable("student")


    //写sql  使用上面注册的临时表
    //自动将sql查询结果转换成一个DataFrame
    val df = sqlContext.sql("select name,age from student")
    df.show()


    //统计班级人数
    val clazzCount = sqlContext.sql("select clazz,count(1) as count from student group by clazz")


    // 将DF的数据保存到hdfs  以json格式保存
    clazzCount
      .write
      //保存数据的方式   追加，覆盖，存在就报错，存在就忽略
      .mode(SaveMode.Overwrite)
      .json("spark/data/clazz_count")


    /**
      * parquet   数据由列名，可以直接通过sparksql读取  ，数据会由一定的压缩，
      * 可以和hive整合
      *
      */
    clazzCount
      .write
      .mode(SaveMode.Overwrite)
      .parquet("spark/data/parquet")


    //读取parquet
    sqlContext.read.parquet("spark/data/parquet").show()


  }
}
